package cn.getech.data.development.controller;

import cn.getech.data.development.config.properties.BdpJobConfig;
import cn.getech.data.development.constant.DataDevelopmentBizExceptionEnum;
import cn.getech.data.development.constant.JobLogTypeEnum;
import cn.getech.data.development.constant.LastRunState;
import cn.getech.data.development.constant.OutHourStatus;
import cn.getech.data.development.dto.JobLinkAndJobNodeDto;
import cn.getech.data.development.dto.JobLinkDto;
import cn.getech.data.development.dto.JobNodeInfoDto;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.model.dto.DownloadLogDto;
import cn.getech.data.development.model.dto.JobLogDto;
import cn.getech.data.development.model.qo.DevOpsQO;
import cn.getech.data.development.model.vo.*;
import cn.getech.data.development.service.*;
import cn.getech.data.development.utils.hadoop.HadoopLogUtils;
import cn.getech.data.development.utils.hadoop.ShellSparkUtils;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.PageUtils;
import cn.getech.data.intelligence.common.utils.PojoUtils;
import cn.getech.data.intelligence.common.utils.R;
import cn.getech.data.intelligence.common.validator.Assert;
import cn.getech.system.center.annotation.SysLog;
import cn.getech.system.center.constant.OperationModulesEnum;
import cn.getech.system.center.constant.OperationTypeEnum;
import cn.getech.system.center.constant.module.ModuleEnum;
import cn.getech.system.center.entity.SysUserEntity;
import cn.getech.system.center.model.vo.SysUserAndEmailVO;
import cn.getech.system.center.service.SysUserService;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.oozie.client.OozieClientException;
import org.joda.time.DateTime;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.text.DecimalFormat;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 任务历史
 *
 * @author zenith
 * @Date 2018-08-27 14:48:51
 */
@Slf4j
@Api(tags = "任务历史")
@RestController
@RequestMapping("/bdp/jobRunHistory")
public class JobNodeRunHistoryController {


    @Autowired
    private JobNodeRunHistoryService jobNodeRunHistoryService;
    @Autowired
    private JobRunHistoryService jobRunHistoryService;
    @Autowired
    private JobNodeInfoService jobNodeInfoService;
    @Autowired
    private JobInfoService jobInfoService;


    @Autowired
    private JobDepRunHistoryService jobDepRunHistoryService;

    @Autowired
    private WorkFlowDepService workFlowDepService;

    @Autowired
    private ProcUserService procUserService;
    @Autowired
    private IWorkMenuJobRelService iWorkMenuJobRelService;

    @Autowired
    private IWorkMenuService workMenuService;

    @Autowired
    private JobLinkService jobLinkService;

    @Autowired
    private BdpJobConfig bdpJobConfig;

    @Autowired
    private DownloadLogService downloadLogService;

    @Autowired
    private WorkFlowLogService  workFlowLogService;

    @Autowired
    private SysUserService sysUserService;

    @Autowired
    private RealTimeTaskMenuService realTimeTaskMenuService;

    @ApiOperation("获取手动和自动的下拉创建人集合")
    @GetMapping("/logUsers")
    public R logUsers(@RequestParam(name = "type") Integer type) {
        //如果是0：手动 1：调度
        List<SysUserEntity> lists = jobNodeRunHistoryService.logUsers(type);
        return R.okWithData(lists);
    }


    @ApiOperation("分页查询离线工作流运维信息")
    @PostMapping("/pageWorkInfo")
    @SysLog(value="分页查询离线工作流运维信息",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.TASK_OPERATION_MODULE_CODE)
    public R pageWorkInfo(@RequestBody DevOpsQO devOpsQO) {
        PageUtils page = workMenuService.queryPage(devOpsQO);
        if (null != page) {
            //封装返回对象
            if(CollectionUtil.isNotEmpty(page.getList())){
                List<JobInfoDevOpsVO> jobInfoDevOpsVOS = jobInfoService.changeBeans((List<WorkMenu>)page.getList());
                page.setList(jobInfoDevOpsVOS);
            }
            return R.okWithPage(page);
        }
        return R.okWithPage(null);
    }

    @ApiOperation("获取所有用户运维的用户信息")
    @GetMapping("/allUsers")
    public R allUsers(@RequestParam(value = "targetFlage")Integer targetFlage) {
        List<Long> userIds = null;
        //0：离线任务
        if(Objects.equals(0,targetFlage)){
            userIds = workMenuService.listUserIds(ShiroUtils.getUserId(),ShiroUtils.getTenantId());
        }
        //1.实时任务
        if(Objects.equals(1,targetFlage)){
            userIds = realTimeTaskMenuService.listUserIds(ShiroUtils.getUserId(),ShiroUtils.getTenantId());
        }
        if(CollectionUtil.isNotEmpty(userIds)){
            List<SysUserEntity> users = sysUserService.getUsers(userIds);
            return R.okWithData(PojoUtils.listConvert(SysUserAndEmailVO.class,users));
        }
        return R.okWithData(null);
    }


    /**
     * 获取列表
     */
    @ApiOperation("获取列表")
    @PostMapping("/list")
//    @SysLog(value="获取列表",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.LOG_MANAGER_MODULE_CODE)
    //@RequiresPermissions("bdp:jobRunHistory:list")
    public R list(@RequestBody Map<String, Object> params){
        //分页获取调度日志
        PageUtils page= workFlowLogService.getLogByPage(params);
        return R.okWithPage(page);
    }



     /**
     * @Description  分页获取调度日志列表，抽取公共方法，（运维中心-离线任务运维、日志管理-手动调度日志/调度作业日志 公共）
     * @Author  chengweiping
     * @Date   2021/1/13 14:44
     */
    private PageUtils  getLogByPage(Map<String, Object> params){
        Integer jobLogType = Integer.parseInt(params.get("job_log_type").toString());
        PageUtils page =null;
        if (jobLogType.equals(JobLogTypeEnum.WORKFLOW.getCode())){
            page = jobRunHistoryService.queryPage(params);
            if (CollectionUtil.isNotEmpty(page.getList())) {
                fillJobRunHistoryInfo((List<JobRunHistory>) page.getList());
            }
        }
        else if (jobLogType.equals(JobLogTypeEnum.WORKFLOWDEP.getCode())){
            page = jobDepRunHistoryService.queryPage(params);
            if (CollectionUtil.isNotEmpty(page.getList())){
                fillJobDepHistoryInfo((List<JobDepHistory>) page.getList());
            }
        }
        return page;

    }


    /**
     * 运维中心-离线运维列表
     */
    @ApiOperation("运维中心-离线运维列表")
    @PostMapping("/allLogList")
    @SysLog(value="运维中心-离线运维列表",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.SYSMODECODE,platformModuleType = ModuleEnum.OPERATION_CENTER_MODULE_CODE)
    public R allLogList(@RequestBody JobLogDto jobLogDto){
        if(jobLogDto==null ){
            jobLogDto=new JobLogDto();
        }

        PageUtils<WorkFlowLogVO> page= workFlowLogService.getWorkFlowLogByPage(jobLogDto);
        return R.okWithPage(page);
    }


    @ApiOperation("运维中心-离线任务列表-下载")
    @RequestMapping(value = "/download", method = RequestMethod.POST)
    @SysLog(value="运维中心-离线运维列表-下载",type= OperationTypeEnum.DOWNLOADCODE, modulesType = OperationModulesEnum.SYSMODECODE,platformModuleType = ModuleEnum.OPERATION_CENTER_MODULE_CODE)
    public void download(HttpServletRequest request,
                              HttpServletResponse response,
                              @RequestBody DownloadLogDto downloadLogDto) throws Exception {

        if(downloadLogDto==null ){
            downloadLogDto=new DownloadLogDto();
           // downloadLogDto.setJob_log_type(0);
        }

        /*if(downloadLogDto.getJob_log_type()==null){
            downloadLogDto.setJob_log_type(0);
        }*/

        request.setCharacterEncoding("UTF-8");
        String prefix=".xlsx";
        String fileName="logList";
        String userAgent = request.getHeader("User-Agent");
        String     downLoadPath = URLDecoder.decode(fileName, "UTF-8");
        System.out.println(downLoadPath);
        try {

            response.setContentType("application/vnd.ms-excel");
            fileName = URLEncoder.encode(fileName, "ISO8859-1");
            response.setHeader("Content-disposition", "attachment;filename=" + fileName + ".xlsx");
            response.setHeader("filename",fileName+prefix);
            JobLogDto  jobLogDto=new JobLogDto();
            BeanUtil.copyProperties(downloadLogDto,jobLogDto);
            jobLogDto.setPage(1);
            //保证查询出来所有数据，目前单个excel可以插入1048576 条数据
            jobLogDto.setLimit(1000000);
            PageUtils<WorkJobAndDepVO> page= workFlowLogService.getWorkFlowLogByPage(jobLogDto);
            List<WorkJobAndDepVO> workFlowLogVOList=page.getList();
            downloadLogService.downloadJobLog(response,workFlowLogVOList);


        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    @ApiOperation("根据工作流依赖的日志信息")
    @GetMapping("/listById/{id}/{jobLogType}")
    //@SysLog(value="根据工作流依赖的日志信息",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.TASK_OPERATION_MODULE_CODE)
    //todo 0925
    //@RequiresPermissions("bdp:jobRunHistory:list")
    public R listById(@PathVariable("id")Long id,@PathVariable("jobLogType")Integer jobLogType){
        if (jobLogType.equals(JobLogTypeEnum.WORKFLOW.getCode())){
            JobRunHistory jobRunHistory = jobRunHistoryService.getById(id);
            List<JobRunHistory> olderList = new ArrayList<>();
            if(null != jobRunHistory){
                if(jobRunHistory.getId() != null && StringUtils.isEmpty(jobRunHistory.getLog()))
                    jobRunHistory.setLog(jobRunHistoryService.getLog(jobRunHistory.getId()));
                olderList.add(jobRunHistory);
                fillJobRunHistoryInfo(olderList);
                //设置创建人
                JobInfo jobInfo= jobInfoService.getById(jobRunHistory.getJobInfoId());
                SysUserEntity sysUserEntity= sysUserService.getById(jobInfo.getCreatePer());
                if(sysUserEntity!=null){
                    olderList.get(0).setCreatePerName(sysUserEntity.getUsername());
                }

            }
            return R.okWithData(olderList);
        }else if (jobLogType.equals(JobLogTypeEnum.WORKFLOWDEP.getCode())){
            JobDepHistory data = jobDepRunHistoryService.getById(id);
            List<JobDepHistory> olderList = new ArrayList<>();
            if(null != data){
                if(data.getId() != null && StringUtils.isEmpty(data.getLog()))
                    data.setLog(jobDepRunHistoryService.getLog(data.getId()));
                olderList.add(data);
                fillJobDepHistoryInfo(olderList);
                //设置创建人
                WorkFlowDep workFlowDep = workFlowDepService.getById(data.getJobDepId());
                WorkMenu workMenu = workMenuService.getById(workFlowDep.getWorkMenuId());
                SysUserEntity sysUserEntity= sysUserService.getById(workMenu.getCreatePer());
                if(sysUserEntity!=null){
                    olderList.get(0).setCreatePerName(sysUserEntity.getUsername());
                }
            }
            return R.okWithData(olderList);
        }
        return R.okWithData(null);
    }




    private void fillJobRunHistoryInfo(List<JobRunHistory> list) {
        DecimalFormat df = new DecimalFormat("0.00");
        Map<Integer, JobRunHistoryVO> map = getEachJobNodeStateIds(list);
        List<Integer> jobIds = list.stream().map(JobRunHistory::getJobInfoId).collect(Collectors.toList());
        //查询这个工作流对应的菜单id
        List<WorkMenuJobRel> workMenuJobRel = iWorkMenuJobRelService.list(new QueryWrapper<WorkMenuJobRel>().eq("work_type",
                1).in("work_flow_id", jobIds));
        Map<Integer, Long> workMap = new HashMap<>();
        if (CollectionUtil.isNotEmpty(workMenuJobRel)) {
            workMap = workMenuJobRel.stream().collect(
                    Collectors.toMap(s -> s.getWorkFlowId().intValue(), WorkMenuJobRel::getWorkMenuId));
        }
        for (JobRunHistory item : list) {
            Integer reRunType = item.getReRunType();
            if(null == reRunType){
                reRunType = 0;
            }
            if(reRunType > 0){
                reRunType = 1;
            }
            item.setReRunType(reRunType);
            item.setStateName(LastRunState.ObjOf(item.getState()).getName());
            item.setJobLogType(JobLogTypeEnum.WORKFLOW.getCode());
            JobRunHistoryVO jobRunHistoryVO = map.get(item.getId());
            item.setNodeStates(jobRunHistoryVO.getNodeStates());
            item.setNodeOutHours(jobRunHistoryVO.getNodeOutHours());
            if (null != item.getCost()) {
                item.setCostSecond(df.format((float) item.getCost() / 1000));
            }
            item.setWorkMenuId(workMap.get(item.getJobInfoId()));
        }
    }
    // todo 需要优化
    private void fillJobDepHistoryInfo(List<JobDepHistory> list) {
        DecimalFormat df = new DecimalFormat("0.00");
        for (JobDepHistory item :list){
            WorkFlowDep workFlowDep = workFlowDepService.getById(item.getJobDepId());
            WorkMenu workMenu = workMenuService.getById(workFlowDep.getWorkMenuId());
            if(null == workMenu){
                item.setDepName("已删除的作业流依赖");
            }else{
                item.setDepName(workMenu.getName());
            }
            Integer reRunType = item.getReRunType();
            if(null == reRunType){
                reRunType = 0;
            }
            if(reRunType > 0){
                reRunType = 1;
            }
            item.setReRunType(reRunType);
            item.setJobLogType(JobLogTypeEnum.WORKFLOWDEP.getCode());
            item.setStateName(LastRunState.ObjOf(item.getState()).getName());
            JobDepRunHistoryVO depRunHistoryVO = new JobDepRunHistoryVO();
            BeanUtils.copyProperties(item,depRunHistoryVO);
            depRunHistoryVO = getEachDepNodeState(item.getJobDepId().intValue(),item.getId(), depRunHistoryVO, item.getOozieJobId());
            item.setNodeStates(depRunHistoryVO.getNodeStates());
            item.setNodeOutHours(depRunHistoryVO.getNodeOutHours());
            if(null != item.getCost()){
                item.setCostSecond(df.format((float)item.getCost()/1000));
            }
            item.setWorkMenuId(workFlowDep.getWorkMenuId());
        }
    }

    /**
     * 删除任务节点日志详情
     */
    @ApiOperation("删除任务节点日志详情")
    @GetMapping("/deleteHistory")
    @SysLog(value="删除任务节点日志详情",type= OperationTypeEnum.DELETECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.TASK_OPERATION_MODULE_CODE)
    //todo 0925
    //@RequiresPermissions("bdp:jobRunHistory:delete")
    public R deleteHistory(@RequestParam Map<String, Object> params){
        Object runId = params.get("runId");
        Object jobLogType = params.get("jobLogType");
        Assert.isNull(runId,"日志编号不能为空");
        Assert.isNull(jobLogType,"日志类型不能为空");
        Integer jobLogTypeCode = Integer.parseInt(jobLogType.toString());
        if (jobLogTypeCode.equals(JobLogTypeEnum.WORKFLOW.getCode())){
           JobRunHistory jobRunHistory =  jobRunHistoryService.getById(Integer.parseInt(runId.toString()));
           if (jobRunHistory==null){
               throw new RRException("日志不存在");
           }
           if (jobRunHistory.getState().equals(LastRunState.RUNNING.getCode())){
               throw new RRException("任务正在运行中，不能进行删除");
           }
           if (!(jobRunHistory.getDepOozieJobId()==null||"".equals(jobRunHistory.getDepOozieJobId()))){
               JobDepHistory jobDepHistory = jobDepRunHistoryService.getOne(new QueryWrapper<JobDepHistory>().eq("oozie_job_id",jobRunHistory.getDepOozieJobId()));
               if (jobDepHistory!=null){
                   throw new RRException("该日志属于编号为"+jobDepHistory.getId()+"的任务依赖日志的子日志，不能被直接删除，如有需要请通过删除对应任务依赖日志来删除该日志");
               }
           }
           JobInfo jobInfo =jobInfoService.getById(jobRunHistory.getJobInfoId());
           if (jobInfo==null){
               throw new RRException("该任务不存在");
           }
           List<Integer> userIds =procUserService.procUserIdList(jobInfo.getProcId());
           if (!userIds.contains(ShiroUtils.getUserId().intValue())){
               throw  new RRException("没有权限删除该日志");
           }
           jobRunHistoryService.removeLogicById(Integer.parseInt(runId.toString()));
           jobNodeRunHistoryService.removeLogicBatch(Collections.singletonList(jobRunHistory.getOozieJobId()));

        }
        else if(jobLogTypeCode.equals(JobLogTypeEnum.WORKFLOWDEP.getCode())){
            JobDepHistory jobDepHistory = jobDepRunHistoryService.getById(Integer.parseInt(runId.toString()));
            if (jobDepHistory==null){
                throw new RRException("日志不存在");
            }
            if (jobDepHistory.getState().equals(LastRunState.RUNNING.getCode())){
                throw new RRException("该任务依赖正在运行中，不能进行删除");
            }
            WorkFlowDep workFlowDep = workFlowDepService.getById(jobDepHistory.getJobDepId());
            if (null==workFlowDep){
                throw new RRException("该工作流依赖不存在");
            }
            List<Integer> procUserIds = procUserService.procUserIdList(workFlowDep.getProcId());
            if (!procUserIds.contains(ShiroUtils.getUserId().intValue())){
                throw  new RRException("没有权限删除该日志");
            }
            jobDepRunHistoryService.removeLogicById(Integer.parseInt(runId.toString()));
            List<JobRunHistory> jobRunHistorys = jobRunHistoryService.list(new QueryWrapper<JobRunHistory>().eq("dep_oozie_job_id",jobDepHistory.getOozieJobId()));
            //jobRunHistoryService.removeLogic(jobDepHistory.getOozieJobId());
            jobRunHistoryService.removeLogic(jobDepHistory);
            if (CollectionUtil.isNotEmpty(jobRunHistorys)) {
                List<String> ozJobIds = jobRunHistorys.stream().map(JobRunHistory::getOozieJobId).collect(Collectors.toList());
                jobNodeRunHistoryService.removeLogicBatch(ozJobIds);
            }

        }
        return R.ok();
    }


//    /**
//     * 任务节点日志详情
//     */
//    @ApiOperation("任务节点日志详情")
//    @GetMapping("/nodeInfo")
//    public R nodeInfo(@RequestParam Map<String, Object> params){
//        Object jobRunId = params.get("jobRunId");
////        Object nodeKey = params.get("nodeKey");
//        if(null == jobRunId){
//            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
//        }
//        //state 0:失败  1：成功 2：执行中 3： 当前版本未执行
//        JobRunHistory jobRunHistory = jobRunHistoryService.getOne(new QueryWrapper<JobRunHistory>().eq("id",params.get("jobRunId")));
//        if(null == jobRunHistory){
//            JobRunHistoryVO jobRunHistoryVO = new JobRunHistoryVO();
//            jobRunHistoryVO.setState(LastRunState.NOTRUN.getCode());
//            jobRunHistoryVO.setLog("当前版本未执行");
//            return R.okWithData(jobRunHistoryVO);
//        }
//        DecimalFormat df = new DecimalFormat("0.00");
//        JobRunHistoryVO jobRunHistoryVO = new JobRunHistoryVO();
//        BeanUtils.copyProperties(jobRunHistory,jobRunHistoryVO);
//        if(null != jobRunHistory.getCost()){
//            jobRunHistoryVO.setCostSecond(df.format((float)jobRunHistory.getCost()/1000));
//        }
//        //根据历史表中查询
//        List<JobNodeRunHistory> jobNodeRunHistorys = jobNodeRunHistoryService.list(new QueryWrapper<JobNodeRunHistory>()
//                .eq("num",jobRunHistory.getNum()).eq("job_info_id", jobRunHistory.getJobInfoId()));
//        if(CollectionUtil.isEmpty(jobNodeRunHistorys)){
//            jobRunHistoryVO.setState(LastRunState.NOTRUN.getCode());
//            jobRunHistoryVO.setLog("当前版本未执行");
//            return R.okWithData(jobRunHistoryVO);
//        }
//        List<JobRunHistoryVO> allNodeKeyStatus = new ArrayList<>();
//        for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistorys) {
//            JobRunHistoryVO jobRunHistoryVO1 = new JobRunHistoryVO();
//            BeanUtils.copyProperties(jobNodeRunHistory,jobRunHistoryVO1);
//            jobRunHistoryVO1.setBeginTime(jobNodeRunHistory.getTime());
//            if(null == jobNodeRunHistory){
//                jobRunHistoryVO1.setLog("当前版本未执行");
//                jobRunHistoryVO1.setState(LastRunState.NOTRUN.getCode());
//            }else {
//                if(StringUtils.isEmpty(jobNodeRunHistory.getLog())){
//                    jobRunHistoryVO1.setLog("执行中");
//                    jobRunHistoryVO1.setState(LastRunState.RUNNING.getCode());
//                }else {
//                    jobRunHistoryVO1.setLog(jobNodeRunHistory.getLog().replace("\n","</br>"));
//                    jobRunHistoryVO1.setState(jobNodeRunHistory.getState());
//                }
//            }
//            if(null != jobNodeRunHistory.getCost()){
//                jobRunHistoryVO1.setCostSecond(df.format((float)jobNodeRunHistory.getCost()/1000));
//            }
//            allNodeKeyStatus.add(jobRunHistoryVO1);
//        }
//        jobRunHistoryVO.setJobNodeHistoryList(allNodeKeyStatus);
//        return R.okWithData(jobRunHistoryVO);
//    }


    /**
     * 任务节点日志轮询查询
     */
    @ApiOperation("任务节点日志轮询查询")
    @GetMapping("/getNodeRunInfo")
    public R getNodeRunInfo(@RequestParam Map<String, Object> params) throws OozieClientException {
        Object jobRunId = params.get("runId");
//        Object nodeKey = params.get("nodeKey");
//        Object startNodeNum = params.get("startNodeNum");
        Object jobNodeInfoId = params.get("jobNodeInfoId");
        Object oozieId = params.get("oozieId");
        if(null == jobNodeInfoId){
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if(null == oozieId){
//            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
            throw new RRException("未执行！");
        }
        JobNodeInfo jobNodeInfo=jobNodeInfoService.getById(Integer.parseInt(jobNodeInfoId.toString()));
        String jobName = "";
        if(null != jobNodeInfo){
            JobInfo jobInfo = jobInfoService.getById(jobNodeInfo.getJobId());
            if(null != jobInfo){
                jobName = jobInfo.getName();
            }
        }
        //state 0:失败  1：成功 2：执行中 3： 当前版本未执行
        JobRunHistory jobRunHistory = null;
        if(StringUtils.isNotEmpty(jobRunId.toString())){
            jobRunHistory = jobRunHistoryService.getOne(new QueryWrapper<JobRunHistory>().eq("id",jobRunId));
        }
        if(null == jobRunHistory){
            JobRunHistoryVO jobRunHistoryVO = new JobRunHistoryVO();
            jobRunHistoryVO.setState(LastRunState.NOTRUN.getCode());
            jobRunHistoryVO.setLog("当前版本未执行");
            JobNodeRunHistory jobNodeRunHistory = new JobNodeRunHistory();
            jobNodeRunHistory.setState(LastRunState.NOTRUN.getCode());
            if (jobNodeInfo!=null){
                jobNodeRunHistory.setName(jobNodeInfo.getName());
            }
            jobNodeRunHistory.setLog("当前版本未执行");
            List<JobNodeRunHistory> jobNodeRunHistories = new ArrayList<>();
            jobNodeRunHistories.add(jobNodeRunHistory);
            jobRunHistoryVO.setJobNodeRunHistories(jobNodeRunHistories);
            jobRunHistoryVO.setName(jobName);
            return R.okWithData(jobRunHistoryVO);
        }
        DecimalFormat df = new DecimalFormat("0.00");
        JobRunHistoryVO jobRunHistoryVO = new JobRunHistoryVO();
        BeanUtils.copyProperties(jobRunHistory,jobRunHistoryVO);
        jobRunHistoryVO.setName(jobName);
        if(null != jobRunHistory.getCost()){
            jobRunHistoryVO.setCostSecond(df.format((float)jobRunHistory.getCost()/1000));
        }
        //根据历史表中查询
        JobNodeRunHistory jobNodeRunHistory = jobNodeRunHistoryService.getOne(new QueryWrapper<JobNodeRunHistory>()
                .eq("job_node_info_id",jobNodeInfoId).eq("oozie_job_id",oozieId));
        if(null == jobNodeRunHistory){
            JobNodeRunHistory jobNodeRunHistory1 = new JobNodeRunHistory();
            List<JobNodeRunHistory> jobNodeRunHistories = new ArrayList<>();
            jobNodeRunHistory1.setState(LastRunState.NOTRUN.getCode());
            jobNodeRunHistory1.setLog("当前版本未执行");
            jobNodeRunHistory1.setCost(null);
            jobNodeRunHistory1.setCostSecond(null);
            if (jobNodeInfo!=null){
                jobNodeRunHistory1.setName(jobNodeInfo.getName());
            }
            jobNodeRunHistories.add(jobNodeRunHistory1);
            jobRunHistoryVO.setJobNodeRunHistories(jobNodeRunHistories);
            getEachJobNodeState(jobRunHistory.getJobInfoId(),Integer.parseInt(jobRunId.toString()),jobRunHistoryVO,oozieId.toString());
            return R.okWithData(jobRunHistoryVO);
        }
        if(null == jobNodeRunHistory.getState() ||  jobNodeRunHistory.getState() == LastRunState.NOTRUN.getCode()){
            List<JobNodeRunHistory> jobNodeRunHistories = new ArrayList<>();
            jobNodeRunHistory.setLog("当前版本未执行");
//            jobRunHistoryVO1.setJenkinsDone(0);
            jobNodeRunHistory.setCost(null);
            jobNodeRunHistory.setCostSecond(null);
            if (jobNodeInfo!=null){
                jobNodeRunHistory.setName(jobNodeInfo.getName());
            }
            jobNodeRunHistories.add(jobNodeRunHistory);
            jobRunHistoryVO.setJobNodeRunHistories(jobNodeRunHistories);
            getEachJobNodeState(jobRunHistory.getJobInfoId(),Integer.parseInt(jobRunId.toString()),jobRunHistoryVO,oozieId.toString());
            return R.okWithData(jobRunHistoryVO);
        }

        if(LastRunState.FAIL.getCode() == jobNodeRunHistory.getState() || LastRunState.SUCCESS.getCode() == jobNodeRunHistory.getState() || LastRunState.STOP.getCode() == jobNodeRunHistory.getState()){
            if(null != jobNodeRunHistory.getCost()){
                jobNodeRunHistory.setCostSecond(df.format((float)jobNodeRunHistory.getCost()/1000));
            }
            if (jobNodeInfo!=null){
                jobNodeRunHistory.setName(jobNodeInfo.getName());
            }
            List<JobNodeRunHistory> jobNodeRunHistories = new ArrayList<>();
            jobNodeRunHistories.add(jobNodeRunHistory);
            jobRunHistoryVO.setJobNodeRunHistories(jobNodeRunHistories);
            wrapLog(jobNodeRunHistories);
            getEachJobNodeState(jobRunHistory.getJobInfoId(),Integer.parseInt(jobRunId.toString()),jobRunHistoryVO,oozieId.toString());
            return R.okWithData(jobRunHistoryVO);
        }


//        JobNodeJenkinsNumRel jobNodeJenkins = jobNodeJenkinsNumRelService.getOne(new QueryWrapper<JobNodeJenkinsNumRel>().eq("job_node_id", jobNodeRunHistory.getJobNodeInfoId()).eq("job_node_num", startNodeNum));
//        if(null == jobNodeJenkins){
//            return R.error("任务运行启动中请稍等...");
//        }
        JobInfo jobInfo = jobInfoService.getById(jobRunHistory.getJobInfoId());
        if(null == jobInfo){
            return R.error("记录任务信息错误");
        }


        List<JobNodeRunHistory> jobNodeRunHistories = new ArrayList<>();
        jobRunHistoryVO.setState(LastRunState.RUNNING.getCode());
        if(null != jobNodeRunHistory.getTime()){
            jobRunHistoryVO.setCostSecond(df.format((float)(DateTime.now().toDate().getTime() - jobNodeRunHistory.getTime().getTime())/1000));
        }
        if (jobNodeInfo!=null){
            jobNodeRunHistory.setName(jobNodeInfo.getName());
        }
        jobRunHistoryVO.setCost(null);
        jobNodeRunHistories.add(jobNodeRunHistory);
        jobRunHistoryVO.setJobNodeRunHistories(jobNodeRunHistories);
        wrapLog(jobNodeRunHistories);
        getEachJobNodeState(jobRunHistory.getJobInfoId(),Integer.parseInt(jobRunId.toString()),jobRunHistoryVO,oozieId.toString());
        return R.okWithData(jobRunHistoryVO);
    }

    private void wrapLog(List<JobNodeRunHistory> jobNodeRunHistories){
        if (CollectionUtil.isNotEmpty(jobNodeRunHistories)) {
            for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistories) {
                if (StringUtils.isNotEmpty(jobNodeRunHistory.getLog())) {
                    try {
                        JSONObject logJson = JSONObject.parseObject(jobNodeRunHistory.getLog());
                        String consoleUrl = logJson.getString("consoleUrl");
                        //"http://bigdata-test-4:8088/proxy/application_1600347239206_35663/"
                        if (StringUtils.isNotEmpty(consoleUrl)) {
                            String applicationId = consoleUrl.split("/")[4];
                            for (String defaultFS : bdpJobConfig.getNamenodestr().split(",")) {
                                    HadoopLogUtils.setDefaultFS(defaultFS);
                                    String  s = HadoopLogUtils.printLog(applicationId, "yarn");
                                if (StringUtils.isNotEmpty(s)) {
                                    jobNodeRunHistory.setFullLog(s);
                                    break;
                                } else {
                                    log.warn("日志解析错误defaultFS:{},applicationId:{},owner:{}", defaultFS, applicationId, "yarn");
                                }
                                s = HadoopLogUtils.printLog(applicationId, "hdfs");
                                if (StringUtils.isNotEmpty(s)) {
                                    jobNodeRunHistory.setFullLog(s);
                                    break;
                                } else {
                                    log.warn("日志解析错误defaultFS:{},applicationId:{},owner:{}", defaultFS, applicationId, "hdfs");
                                }
                                s = HadoopLogUtils.printLog(applicationId, "spark");
                                if (StringUtils.isNotEmpty(s)) {
                                    jobNodeRunHistory.setFullLog(s);
                                    break;
                                } else {
                                    log.warn("日志解析错误defaultFS:{},applicationId:{},owner:{}", defaultFS, applicationId, "spark");
                                }
                            }
                        }
                    } catch (Exception e) {
                        log.warn("无效日志");
                    }

                }
            }
        }
    }
    public Map<Integer,JobRunHistoryVO> getEachJobNodeStateIds(List<JobRunHistory> list){
        Map<Integer, JobRunHistoryVO> jobRunHistoryVOMap = new HashMap<>();
        List<Integer> jobId = list.stream().map(JobRunHistory::getJobInfoId).collect(Collectors.toList());
        List<Integer> jobRunIds = list.stream().map(JobRunHistory::getId).collect(Collectors.toList());
        List<String> ozId = list.stream().map(JobRunHistory::getOozieJobId).collect(Collectors.toList());
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoService.list(new QueryWrapper<JobNodeInfo>().in("job_id", jobId));
        Map<Integer, List<JobNodeInfo>> jobNodeInfosMap = new HashMap<>();
        if (CollectionUtil.isNotEmpty(jobNodeInfos)) {
            jobNodeInfosMap = jobNodeInfos.stream().collect(Collectors.groupingBy(JobNodeInfo::getJobId));
        }
        Map<Integer, List<JobRunHistory>> jobRunHistoryMap = new HashMap<>();
        List<JobRunHistory> jobRunHistory = jobRunHistoryService.list(new QueryWrapper<JobRunHistory>().in("id", jobRunIds));
        if (CollectionUtil.isNotEmpty(jobRunHistory)) {
            jobRunHistoryMap = jobRunHistory.stream().collect(Collectors.groupingBy(JobRunHistory::getId));
        }
        Map<String, List<JobNodeRunHistory>> jobNodeRunHistoryListMap = new HashMap<>();
        List<JobNodeRunHistory> jobNodeRunHistoryList = jobNodeRunHistoryService.list(new QueryWrapper<JobNodeRunHistory>()
                .in("oozie_job_id", ozId));
        if (CollectionUtil.isNotEmpty(jobNodeRunHistoryList)) {
            jobNodeRunHistoryListMap = jobNodeRunHistoryList.stream().collect(Collectors.groupingBy(JobNodeRunHistory::getOozieJobId));
        }
        for (JobRunHistory runHistory : list) {
            JobRunHistoryVO jobRunHistoryVO = new JobRunHistoryVO();
            BeanUtils.copyProperties(runHistory,jobRunHistoryVO);
            if (CollectionUtil.isEmpty(jobNodeRunHistoryListMap.get(runHistory.getOozieJobId()))
                    && CollectionUtil.isEmpty(jobRunHistoryMap.get(runHistory.getId()))) {
                jobRunHistoryVO.setState(LastRunState.NOTRUN.getCode());
                jobRunHistoryVO.setLog("当前版本未执行");
            } else {
                Map<String, Integer> nodeStates = new HashMap<>();
                //增加每个节点的超时状态记录
                Map<String, Integer> nodeOutHours = new HashMap<>();
                List<JobNodeRunHistory> ozJobList = jobNodeRunHistoryListMap.get(runHistory.getOozieJobId());
                if (CollectionUtil.isNotEmpty(ozJobList)) {
                    for (JobNodeRunHistory jobNodeRunHistory : ozJobList) {
                        if (null == jobNodeRunHistory &&
                                CollectionUtil.isEmpty(jobRunHistoryMap.get(runHistory.getReRunNodeId()))
                        ) {
                            nodeStates.put(jobNodeRunHistory.getJobNodeInfoId().toString(), LastRunState.NOTRUN.getCode());
                            //设置节点超时状态
                            nodeOutHours.put(jobNodeRunHistory.getJobNodeInfoId().toString(), OutHourStatus.NON_OUT.getStatus());
                        } else {
                            nodeStates.put(jobNodeRunHistory.getJobNodeInfoId().toString(), jobNodeRunHistory.getState());
                            //设置节点超时状态
                            nodeOutHours.put(jobNodeRunHistory.getJobNodeInfoId().toString(),jobNodeRunHistory.getIsOutHours()!=null?jobNodeRunHistory.getIsOutHours(): OutHourStatus.NON_OUT.getStatus());
                        }
                    }
                }
                List<JobNodeInfo> nodeNodeInfo = jobNodeInfosMap.get(runHistory.getJobInfoId());
                if (CollectionUtil.isNotEmpty(nodeNodeInfo)) {
                    Set<String> nodeKeys = nodeNodeInfo.stream().map(s -> s.getId().toString()).collect(Collectors.toSet());
                    //将未查询到的node状态赋值为未运行
                    nodeKeys.removeAll(nodeStates.keySet());
                    for (String s : nodeKeys
                    ) {
                        nodeStates.put(s, LastRunState.NOTRUN.getCode());
                        //设置节点超时状态
                        nodeOutHours.put(s,OutHourStatus.NON_OUT.getStatus());
                    }
                    jobRunHistoryVO.setNodeStates(nodeStates);
                    jobRunHistoryVO.setNodeOutHours(nodeOutHours);
                }
            }
            jobRunHistoryVOMap.put(runHistory.getId(), jobRunHistoryVO);
        }
        return jobRunHistoryVOMap;
    }


    public JobRunHistoryVO getEachJobNodeState(Integer jobId, Integer jobRunId, JobRunHistoryVO jobRunHistoryVO,String oozieId){
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoService.list(new QueryWrapper<JobNodeInfo>().eq("job_id",jobId));
        Set<String> nodeKeys = new HashSet<>();
        for (JobNodeInfo jobNodeInfo :  jobNodeInfos
             ) {
            nodeKeys.add(jobNodeInfo.getId().toString());
        }
        JobRunHistory jobRunHistory = jobRunHistoryService.getOne(new QueryWrapper<JobRunHistory>().eq("id", jobRunId));
        List<JobNodeRunHistory> jobNodeRunHistorys = jobNodeRunHistoryService.list(new QueryWrapper<JobNodeRunHistory>()
                .eq("oozie_job_id",oozieId));
        if(CollectionUtil.isEmpty(jobNodeRunHistorys) && jobRunHistory == null){
            jobRunHistoryVO.setState(LastRunState.NOTRUN.getCode());
            jobRunHistoryVO.setLog("当前版本未执行");
            return jobRunHistoryVO;
        }
        Map<String,Integer> nodeStates = new HashMap<>();
        //增加每个节点的超时状态记录
        Map<String, Integer> nodeOutHours = new HashMap<>();
        for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistorys) {
            if(null == jobNodeRunHistory && jobRunHistory == null){
                nodeStates.put(jobNodeRunHistory.getJobNodeInfoId().toString(),LastRunState.NOTRUN.getCode());
                //设置节点超时状态
                nodeOutHours.put(jobNodeRunHistory.getJobNodeInfoId().toString(), OutHourStatus.NON_OUT.getStatus());
            }else {
                    nodeStates.put(jobNodeRunHistory.getJobNodeInfoId().toString(),jobNodeRunHistory.getState());
                //设置节点超时状态
                nodeOutHours.put(jobNodeRunHistory.getJobNodeInfoId().toString(),jobNodeRunHistory.getIsOutHours()!=null?jobNodeRunHistory.getIsOutHours(): OutHourStatus.NON_OUT.getStatus());
            }
        }
        //将未查询到的node状态赋值为未运行
        nodeKeys.removeAll(nodeStates.keySet());
        for (String s : nodeKeys
             ) {
            nodeStates.put(s,LastRunState.NOTRUN.getCode());
            //设置节点超时状态
            nodeOutHours.put(s, OutHourStatus.NON_OUT.getStatus());
        }
        jobRunHistoryVO.setNodeStates(nodeStates);
        jobRunHistoryVO.setNodeOutHours(nodeOutHours);
        return jobRunHistoryVO;
    }


    public JobDepRunHistoryVO getEachDepNodeState(Integer jobDepId, Integer jobDepRunId, JobDepRunHistoryVO jobDepRunHistoryVO,String depJobOozieId){
        WorkFlowDep workFlowDep =workFlowDepService.getById(jobDepId);

        List<WorkMenuJobRel> workMenuJobRels = iWorkMenuJobRelService.list(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_menu_id", workFlowDep.getWorkMenuId()));
        Set<String> jobIds = new HashSet<>();
        for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
            jobIds.add(workMenuJobRel.getWorkFlowId().toString());
        }
        List<Integer> jobIdList = new ArrayList<>();
        jobIds.forEach(j->{
            jobIdList.add(Integer.parseInt(j));
        });

//        List<JobInfo> jobInfos = jobInfoService.list(new QueryWrapper<JobInfo>().in("id",jobIdList));

        JobDepHistory jobDepHistory = jobDepRunHistoryService.getOne(new QueryWrapper<JobDepHistory>().eq("id", jobDepRunId));
        List<JobRunHistory> jobRunHistories = jobRunHistoryService.list(new QueryWrapper<JobRunHistory>()
                .eq("dep_oozie_job_id",depJobOozieId));
        if(CollectionUtil.isEmpty(jobRunHistories) && jobDepHistory == null){
            jobDepRunHistoryVO.setState(LastRunState.NOTRUN.getCode());
            jobDepRunHistoryVO.setLog("当前版本未执行");
            return jobDepRunHistoryVO;
        }
        Map<String,Integer> nodeStates = new HashMap<>();
        //增加每个作业流的超时状态记录
        Map<String, Integer> nodeOutHours = new HashMap<>();
        for (JobRunHistory jobRunHistory : jobRunHistories) {
            if(null == jobRunHistory && jobDepHistory == null){
                nodeStates.put(jobRunHistory.getJobInfoId().toString(),LastRunState.NOTRUN.getCode());
                //设置每个作业流超时状态
                nodeOutHours.put(jobRunHistory.getJobInfoId().toString(), OutHourStatus.NON_OUT.getStatus());
            }else {
                nodeStates.put(jobRunHistory.getJobInfoId().toString(),jobRunHistory.getState());
                //设置每个作业流超时状态
                nodeOutHours.put(jobRunHistory.getJobInfoId().toString(),jobRunHistory.getIsOutHours()!=null?jobRunHistory.getIsOutHours(): OutHourStatus.NON_OUT.getStatus());

            }
        }
        //将未查询到的node状态赋值为未运行
        jobIds.removeAll(nodeStates.keySet());
        for (String s : jobIds
        ) {
            nodeStates.put(s,LastRunState.NOTRUN.getCode());
            //设置每个作业流超时状态
            nodeOutHours.put(s, OutHourStatus.NON_OUT.getStatus());
        }
        jobDepRunHistoryVO.setNodeStates(nodeStates);
        jobDepRunHistoryVO.setNodeOutHours(nodeOutHours);
        return jobDepRunHistoryVO;
    }

    /**
     * 任务依赖节点日志轮询查询
     */
    @ApiOperation("任务依赖节点日志轮询查询")
    @GetMapping("/getDepNodeRunInfo")
    public R getDepNodeRunInfo(@RequestParam Map<String, Object> params) throws OozieClientException {
        Object runId = params.get("runId");
//        Object nodeKey = params.get("nodeKey");
//        Object startNodeNum = params.get("startNodeNum");
        Object jobInfoId = params.get("jobInfoId");
        Object oozieId = params.get("oozieId");
        if(null == jobInfoId){
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if(null == oozieId){
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if (null==runId){
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        JobInfo jobInfo =jobInfoService.getById(Integer.parseInt(jobInfoId.toString()));
        //查询对应的工作流的名字
        if(null == jobInfo){
            throw new RRException("工作流不存在！");
        }

        //查询到这个工作流对应的菜单信息
        String workFlowLayName = "";
        WorkMenuJobRel workMenuJobRel1 = iWorkMenuJobRelService.getOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type",2).eq("work_flow_id",jobInfoId));
        if(null != workMenuJobRel1){
            WorkMenu workMenu = workMenuService.getById(workMenuJobRel1.getWorkMenuId());
            if(null != workMenu){
                workFlowLayName = workMenu.getName();
            }
        }
        //state 0:失败  1：成功 2：执行中 3： 当前版本未执行
        JobDepHistory jobDepHistory = jobDepRunHistoryService.getOne(new QueryWrapper<JobDepHistory>().eq("id",runId));
        if(null == jobDepHistory){
            JobDepRunHistoryVO jobDepRunHistoryVO = new JobDepRunHistoryVO();
            jobDepRunHistoryVO.setState(LastRunState.NOTRUN.getCode());
            jobDepRunHistoryVO.setLog("当前版本未执行");
            List<JobRunHistory> list = new ArrayList<>();
            JobRunHistory jobRunHistory = new JobRunHistory();
            jobRunHistory.setState(LastRunState.NOTRUN.getCode());
            jobRunHistory.setLog("当前版本未执行");
            if (jobInfo!=null){
                jobRunHistory.setName(jobInfo.getName());
            }
            list.add(jobRunHistory);
            jobDepRunHistoryVO.setJobHistoryList(list);
            jobDepRunHistoryVO.setName(workFlowLayName);
            return R.okWithData(jobDepRunHistoryVO);
        }
        DecimalFormat df = new DecimalFormat("0.00");
        JobDepRunHistoryVO jobDepRunHistoryVO = new JobDepRunHistoryVO();
        BeanUtils.copyProperties(jobDepHistory,jobDepRunHistoryVO);
        if(null != jobDepHistory.getCost()){
            jobDepRunHistoryVO.setCostSecond(df.format((float)jobDepHistory.getCost()/1000));
        }
        jobDepRunHistoryVO.setName(workFlowLayName);
        //根据历史表中查询
        JobRunHistory jobRunHistory = jobRunHistoryService.getOne(new QueryWrapper<JobRunHistory>()
                .eq("job_info_id",jobInfoId).eq("dep_oozie_job_id",oozieId));

        if(null == jobRunHistory){
            jobRunHistory  = new JobRunHistory();
            //查询当前对应的配置的节点虚拟进去
            List<JobNodeInfo> list = jobNodeInfoService.list(new QueryWrapper<JobNodeInfo>().eq("job_id", jobInfoId).isNotNull("pos_x").isNotNull("pos_y"));
            WorkMenuJobRel workMenuJobRel = iWorkMenuJobRelService.getOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type",1).eq("work_flow_id",jobInfoId));
            JobLinkAndJobNodeDto jobLinkAndJobNodeDto = new JobLinkAndJobNodeDto();
            if(null != workMenuJobRel){
                List<JobLink> linkLists = jobLinkService.list(new QueryWrapper<JobLink>().eq("linke_type",0).eq("menu_id",workMenuJobRel.getWorkMenuId()));
                jobLinkAndJobNodeDto.setEdges(PojoUtils.listConvert(JobLinkDto.class,linkLists));
                jobLinkAndJobNodeDto.setNodes(PojoUtils.listConvert(JobNodeInfoDto.class,list));
            }
            jobRunHistory.setNodeParams(JSONObject.toJSONString(jobLinkAndJobNodeDto));
            List<JobRunHistory> jobRunHistories = new ArrayList<>();
            jobRunHistory.setState(LastRunState.NOTRUN.getCode());
            jobRunHistory.setStateName(LastRunState.NOTRUN.getName());
            jobRunHistory.setLog("当前版本未执行");
            jobRunHistory.setCost(null);
            jobRunHistory.setCostSecond(null);
            if (jobInfo!=null){
                jobRunHistory.setName(jobInfo.getName());
            }
            jobRunHistory.setTime(jobRunHistory.getBeginTime());
            packJobRunHistoryNodeList(jobRunHistory,jobInfoId,null);
            jobRunHistories.add(jobRunHistory);
            jobDepRunHistoryVO.setJobHistoryList(jobRunHistories);
            SysUserEntity sysUserEntity= sysUserService.getById(jobInfo.getCreatePer());
            if(sysUserEntity!=null){
                jobDepRunHistoryVO.setCreatePerName(sysUserEntity.getUsername());
                jobRunHistory.setCreatePerName(sysUserEntity.getUsername());
            }
            getEachDepNodeState(jobDepHistory.getJobDepId().intValue(),Integer.parseInt(runId.toString()),jobDepRunHistoryVO,oozieId.toString());
            return R.okWithData(jobDepRunHistoryVO);
        }
        if(null == jobRunHistory.getState() || jobRunHistory.getState() == LastRunState.NOTRUN.getCode()){
            List<JobRunHistory> jobRunHistories = new ArrayList<>();
            jobRunHistory.setLog("当前版本未执行");
//            jobDepRunHistoryVO1.setJenkinsDone(0);
            jobRunHistory.setCost(null);
            jobRunHistory.setCostSecond(null);
            if (jobInfo!=null){
                SysUserEntity sysUserEntity= sysUserService.getById(jobInfo.getCreatePer());
                if(sysUserEntity!=null){
                    jobDepRunHistoryVO.setCreatePerName(sysUserEntity.getUsername());
                }
                jobRunHistory.setName(jobInfo.getName());
            }
            jobRunHistory.setTime(jobRunHistory.getBeginTime());
            jobRunHistory.setStateName(LastRunState.NOTRUN.getName());
            packJobRunHistoryNodeList(jobRunHistory,jobInfoId,jobRunHistory.getOozieJobId());
            jobRunHistories.add(jobRunHistory);
            jobDepRunHistoryVO.setJobHistoryList(jobRunHistories);
            getEachDepNodeState(jobDepHistory.getJobDepId().intValue(),Integer.parseInt(runId.toString()),jobDepRunHistoryVO,oozieId.toString());
            return R.okWithData(jobDepRunHistoryVO);
        }

        if(LastRunState.FAIL.getCode() == jobRunHistory.getState() || LastRunState.SUCCESS.getCode() == jobRunHistory.getState() || LastRunState.STOP.getCode() == jobRunHistory.getState()){
            List<JobRunHistory> jobRunHistories = new ArrayList<>();
            if (null!=jobRunHistory.getId()&& StringUtils.isNotEmpty(jobRunHistory.getLog())){
                jobRunHistory.setLog(jobRunHistory.getLog().replace("\n","</br>"));
            }else{
                jobRunHistory.setLog(jobRunHistoryService.getLog(jobRunHistory.getId()));
            }
            if(null != jobRunHistory.getCost()){
                jobRunHistory.setCostSecond(df.format((float)jobRunHistory.getCost()/1000));
            }
            jobRunHistory.setCost(null);
            if (jobInfo!=null){
                jobRunHistory.setName(jobInfo.getName());
                SysUserEntity sysUserEntity= sysUserService.getById(jobInfo.getCreatePer());
                if(sysUserEntity!=null){
                    jobDepRunHistoryVO.setCreatePerName(sysUserEntity.getUsername());
                    jobRunHistory.setCreatePerName(sysUserEntity.getUsername());
                }
            }
            jobRunHistory.setTime(jobRunHistory.getBeginTime());
            jobRunHistory.setStateName(LastRunState.ObjOf(jobRunHistory.getState()).getName());
            packJobRunHistoryNodeList(jobRunHistory,jobInfoId,jobRunHistory.getOozieJobId());
            jobRunHistories.add(jobRunHistory);
            jobDepRunHistoryVO.setJobHistoryList(jobRunHistories);
            getEachDepNodeState(jobDepHistory.getJobDepId().intValue(),Integer.parseInt(runId.toString()),jobDepRunHistoryVO,oozieId.toString());
            return R.okWithData(jobDepRunHistoryVO);
        }



        jobRunHistory.setTime(jobRunHistory.getBeginTime());
        JobRunHistory jobRunHistory1 = new JobRunHistory();
        BeanUtil.copyProperties(jobRunHistory,jobRunHistory1);
        List<JobRunHistory> jobRunHistories = new ArrayList<>();
        jobRunHistory1.setState(LastRunState.RUNNING.getCode());
        jobRunHistory1.setStateName(LastRunState.RUNNING.getName());
        if (jobInfo!=null){
            jobRunHistory.setName(jobInfo.getName());
            SysUserEntity sysUserEntity= sysUserService.getById(jobInfo.getCreatePer());
            if(sysUserEntity!=null){
                jobDepRunHistoryVO.setCreatePerName(sysUserEntity.getUsername());
                jobRunHistory1.setCreatePerName(sysUserEntity.getUsername());
                jobRunHistory1.setName(jobInfo.getName());
            }
        }
        if(null != jobRunHistory.getBeginTime()){
            jobRunHistory1.setCostSecond(df.format((float)(DateTime.now().toDate().getTime() - jobRunHistory.getBeginTime().getTime())/1000));
        }
        //对应的日志
        if(null != jobRunHistory && null != jobRunHistory.getId() && StringUtils.isEmpty(jobRunHistory.getLog())){
            jobRunHistory1.setLog(jobRunHistoryService.getLog(jobRunHistory.getId()));
        }
        jobRunHistory1.setCost(null);
        packJobRunHistoryNodeList(jobRunHistory1,jobInfoId,jobRunHistory1.getOozieJobId());
        jobRunHistories.add(jobRunHistory1);
        jobDepRunHistoryVO.setJobHistoryList(jobRunHistories);
        getEachDepNodeState(jobDepHistory.getJobDepId().intValue(),Integer.parseInt(runId.toString()),jobDepRunHistoryVO,oozieId.toString());
        return R.okWithData(jobDepRunHistoryVO);
    }

    private void packJobRunHistoryNodeList(JobRunHistory jobRunHistory1, Object jobInfoId, String oozieJobId) {
        //查询到对应的节点关系图
        if(StringUtils.isNotEmpty(jobRunHistory1.getNodeParams())){
            try {
                JSONObject jsonObject = JSONObject.parseObject(jobRunHistory1.getNodeParams());
                if(null != jsonObject){
                    Object nodes = jsonObject.get("nodes");
                    if(null != nodes){
                        JSONArray  jsonArray = JSONArray.parseArray(nodes.toString());
                        if(null != jsonArray){
                            //遍历每一个节点，并且和oozieId查询到对应的节点的状态数据
                            Map<String,Integer> nodeStates = new HashMap<>();
                            //增加超时状态记录
                            Map<String, Integer> nodeOutHours = new HashMap<>();
                            for (Object o : jsonArray) {
                                if(null != o){
                                    JSONObject obj = JSONObject.parseObject(o.toString());
                                    if(null != obj){
                                        Object id = obj.get("id");
                                        if(null != id){
                                            Integer status = LastRunState.NOTRUN.getCode();
                                            Integer isOutHours= OutHourStatus.NON_OUT.getStatus();
                                            if(StringUtils.isNotEmpty(oozieJobId)){
                                                JobNodeRunHistory one = jobNodeRunHistoryService.getOne(new QueryWrapper<JobNodeRunHistory>().eq("job_node_info_id", id).eq("oozie_job_id", oozieJobId));
                                                if(null != one){
                                                    status = one.getState();
                                                    isOutHours=one.getIsOutHours();
                                                }
                                            }
                                            nodeStates.put(id.toString(),status);
                                            nodeOutHours.put(id.toString(),isOutHours);
                                        }
                                    }
                                }
                            }
                            jobRunHistory1.setNodeStates(nodeStates);
                            jobRunHistory1.setNodeOutHours(nodeOutHours);
                        }
                    }
                }
            }catch (Exception e){
                log.error("解析日志信息异常！error:{}",e.getMessage());
            }
        }else {
            //查询到最近的节点关系图
            List<JobNodeInfo> jobNodeInfos = jobNodeInfoService.list(new QueryWrapper<JobNodeInfo>()
                    .eq("job_id", jobInfoId)
                    .isNotNull("pos_x")
                    .isNotNull("pos_y"));
            if(CollectionUtil.isNotEmpty(jobNodeInfos)){
                Map<String,Integer> nodeStates = new HashMap<>();
                for (JobNodeInfo jobNodeInfo : jobNodeInfos) {
                    Integer status = LastRunState.NOTRUN.getCode();
                    if(StringUtils.isNotEmpty(oozieJobId)){
                        JobNodeRunHistory one = jobNodeRunHistoryService.getOne(new QueryWrapper<JobNodeRunHistory>().eq("job_node_info_id", jobNodeInfo.getId()).eq("oozie_job_id", oozieJobId));
                        if(null != one){
                            status = one.getState();
                        }
                    }
                    nodeStates.put(jobNodeInfo.getId().toString(),status);
                }
                jobRunHistory1.setNodeStates(nodeStates);
            }
        }
    }
}
